package chen.bupt.vsmtest;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import chen.bupt.constant.Constants;
import chen.bupt.text.TextUtils;
import chen.bupt.util.HDFSFileUtils;

/**
 * 用来获取最后一个月的主题贴. 如果用户author已经进行过了评论， 那么标记为1 否则最后输出标记为0
 * 
 * @author forrest
 * 
 */
public class VSMTest {
	private static String author = "ballard";
	private static String inputPath = Constants.SOURCE_INPUT;
	private static String outputPath = "/user/chenlingpeng/smth/user/last_month_topic_";

	public static class Map1 extends Mapper<LongWritable, Text, Text, Text> {
		private static Text t = new Text("key");

		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			if(!HDFSFileUtils.isInputFile(context))return;
			String[] str = TextUtils.splitTextToCont(value.toString(), ";", 7);
			String[] info = str[0].split(";");
			if (info[6].startsWith("2011-08")) {
				if (info[5].equals("ballard")) {
					// 说明不是主题贴
					if (!info[2].equals(info[3])) {
						context.write(t, new Text("1|" + info[4]+"_"+info[1]));
					}
				} else if (info[2].equals(info[4]) || info[4].equals("0")) {
					context.write(t, new Text("0|" + info[2]+"_"+info[1] + "|" + str[1]));
				}
			}
		}
	}

	public static class Reduce1 extends Reducer<Text, Text, Text, NullWritable> {
		public void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			int sum = 0;
			int count = 0;
			// 保存主题贴及内容
			Map<String, String> map0 = new HashMap<String, String>();
			// 保存用户回复过的主题贴
			Set<String> set = new HashSet<String>();
			for (Text value : values) {
				sum++;
				if (value.toString().startsWith("0")) {
					String tmp = value.toString().substring(2);
					int ind = tmp.indexOf("|");
					map0.put(tmp.substring(0, ind), tmp.substring(ind + 1));
					count++;
				} else {
					set.add(value.toString().substring(2));
				}
			}
			context.write(new Text("number:" + sum), NullWritable.get());
			context.write(new Text("number:" + count), NullWritable.get());
			for (String k : map0.keySet()) {
				if (set.contains(k)) {
					context.write(new Text("1|" + map0.get(k)), NullWritable
							.get());
				} else {
					context.write(new Text("0|" + map0.get(k)), NullWritable
							.get());
				}
			}
		}
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		conf.set("mapred.task.timeout", "0");
		Job job = new Job(conf, "VSMTest");
		job.setJarByClass(VSMTest.class);
		job.setMapperClass(Map1.class);
		job.setReducerClass(Reduce1.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		Path input = new Path(inputPath);
		Path output = new Path(outputPath + author);
		HDFSFileUtils.deleteFile(output, conf);
		FileInputFormat.addInputPath(job, input);
		FileOutputFormat.setOutputPath(job, output);
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

}
